package com.fdb.smcts.mq.client.consumer;

import java.util.Map;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragelyByCircle;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fdb.smcts.mq.handle.MQMessageProcessor;
import com.fdb.smcts.mq.listener.MessageListenerForConcurrently;

/**
 * @description:消息接受客户端处理类
 * @date:2019年2月12日 上午9:21:35
 * @author:wangyk3
 */
public class MQPushConsumerMessageClient {
	private DefaultMQPushConsumer consumer;
	private String namesrvAddr;
	private String consumerGroup;
	private String messageListener;
	private Map<String, MQMessageProcessor> handlermap;
	private static Logger logger = LoggerFactory.getLogger(MQPushConsumerMessageClient.class);

	/**
	 * @description:初始化消息接受客户端
	 * @throws InterruptedException
	 * @throws MQClientException
	 * @date:2019年2月12日 上午9:22:17
	 * @author:wangyk3
	 */
	protected void init() throws InterruptedException, MQClientException {
		logger.info("Consumer initialize!!" + consumerGroup + namesrvAddr);
		consumer = new DefaultMQPushConsumer(this.consumerGroup);
		/** 设置namesrv **/
		consumer.setNamesrvAddr(this.namesrvAddr);
		consumer.setInstanceName(String.valueOf(System.currentTimeMillis()));
		/** 订阅主题，仅对新的消费者有效 **/
		subscribe();
		/** 定义消费位置，仅对新的消费者有效 **/
		consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
		/** 设置集群模式 **/
		consumer.setMessageModel(MessageModel.CLUSTERING);
		/** 设置consumeQueue分配算法 **/
		consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragelyByCircle());
		/** 根据consumer类型判断监听类型 **/
		if ("CONCURRENTLY".equals(this.messageListener)) {
			MessageListenerForConcurrently messageListener = new MessageListenerForConcurrently();
			messageListener.setHandlermap(handlermap);
			logger.info("这是一个普通RocketMQ消费！ type:" + this.messageListener);
			consumer.registerMessageListener(messageListener);
		}
		consumer.start();
		logger.info("Consumer start success!");
	}

	/**
	 * s * @description:关闭消息接受客户端
	 * 
	 * @date:2019年2月12日 上午9:22:55
	 * @author:wangyk3
	 */
	public void destroy() {
		consumer.shutdown();
	}

	/**
	 * @description:订阅信息
	 * @date:2019年2月22日 下午3:33:17
	 * @author:wangyk3
	 */
	public void subscribe() {
		for (String topic : handlermap.keySet()) {
			MQMessageProcessor handle = handlermap.get(topic);
			if (handle != null) {
				try {
					consumer.subscribe(topic, "*");
					logger.info("consume subscribe topic" + topic + "success", "*");
				} catch (MQClientException e) {
					logger.error("consume subscribe topic" + topic + "failed", e);
				}
			}
		}
	}

	public DefaultMQPushConsumer getConsumer() {
		return consumer;
	}

	public void setConsumer(DefaultMQPushConsumer consumer) {
		this.consumer = consumer;
	}

	public String getNamesrvAddr() {
		return namesrvAddr;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public String getConsumerGroup() {
		return consumerGroup;
	}

	public void setConsumerGroup(String consumerGroup) {
		this.consumerGroup = consumerGroup;
	}

	public String getMessageListener() {
		return messageListener;
	}

	public void setMessageListener(String messageListener) {
		this.messageListener = messageListener;
	}

	public Map<String, MQMessageProcessor> getHandlermap() {
		return handlermap;
	}

	public void setHandlermap(Map<String, MQMessageProcessor> handlermap) {
		this.handlermap = handlermap;
	}

}
